package stai6271;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.List;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

public class CClient
{
    public static void main(String[] args)
    throws TException, InvalidRequestException, UnavailableException, UnsupportedEncodingException, NotFoundException, TimedOutException
    {
        TTransport tr = new TFramedTransport(new TSocket("198.199.81.128", 9160));
        TProtocol proto = new TBinaryProtocol(tr);
        Cassandra.Client client = new Cassandra.Client(proto);
        tr.open();

/**        String key_user_id = "bobbyjo";

        // insert data
       long timestamp = System.currentTimeMillis();
        client.set_keyspace("DEMO");      
        ColumnParent parent = new ColumnParent("users");*/
        

        
       
        client.set_keyspace("testbed"); 
        ColumnParent parent = new ColumnParent("users");
      

        // insert data
        long timestamp = System.currentTimeMillis();
        String key_user_id = "bicycle0";     
               
        Column lastActCol = new Column(toByteBuffer("LastActivity"));
        lastActCol.setValue(toByteBuffer("2011-04-21T19:56:52.790"));
        lastActCol.setTimestamp(timestamp);
        client.insert(toByteBuffer(key_user_id), parent, lastActCol, ConsistencyLevel.ONE);
        
        Column body = new Column(toByteBuffer("Body"));
        body.setValue(toByteBuffer("<p>I keep losing pressure in my tires, and among other things, I'm looking at ..."));
        body.setTimestamp(timestamp);
        client.insert(toByteBuffer(key_user_id), parent, body, ConsistencyLevel.ONE);
        
        Column ID = new Column(toByteBuffer("Id"));
        ID.setValue(toByteBuffer("1"));
        ID.setTimestamp(timestamp);
        client.insert(toByteBuffer(key_user_id), parent, ID, ConsistencyLevel.ONE);
        
        Column score = new Column(toByteBuffer("Score"));
        score.setValue(toByteBuffer("20"));
        score.setTimestamp(timestamp);
        client.insert(toByteBuffer(key_user_id), parent, score, ConsistencyLevel.ONE);
        
/**        Column nameColumn = new Column(toByteBuffer("fullname"));
        nameColumn.setValue(toByteBuffer("RAVI"));
        nameColumn.setTimestamp(timestamp);
        client.insert(toByteBuffer(key_user_id), parent, nameColumn, ConsistencyLevel.ONE);
        
        Column emailColumn = new Column(toByteBuffer("email"));
        emailColumn.setValue(toByteBuffer("ravi@ou.edu"));
        emailColumn.setTimestamp(timestamp);
        client.insert(toByteBuffer(key_user_id), parent, emailColumn, ConsistencyLevel.ONE);
        
        Column stateColumn = new Column(toByteBuffer("state"));
        stateColumn.setValue(toByteBuffer("OK"));
        stateColumn.setTimestamp(timestamp);
        client.insert(toByteBuffer(key_user_id), parent, stateColumn, ConsistencyLevel.ONE);
        
        Column genderColumn = new Column(toByteBuffer("gender"));
        genderColumn.setValue(toByteBuffer("M"));
        genderColumn.setTimestamp(timestamp);
        client.insert(toByteBuffer(key_user_id), parent, genderColumn, ConsistencyLevel.ONE);
        
        Column byearColumn = new Column(toByteBuffer("birth_year"));
        byearColumn.setValue(toByteBuffer("1985"));
        byearColumn.setTimestamp(timestamp);
        client.insert(toByteBuffer(key_user_id), parent, byearColumn, ConsistencyLevel.ONE);*/

        ColumnPath path = new ColumnPath("users");

        // read single column
        path.setColumn(toByteBuffer("Id"));
        System.out.println("ID: " + client.get(toByteBuffer(key_user_id), path, ConsistencyLevel.ONE));

        // read entire row
        SlicePredicate predicate = new SlicePredicate();
        SliceRange sliceRange = new SliceRange(toByteBuffer(""), toByteBuffer(""), false, 10);
        predicate.setSlice_range(sliceRange);
        
        List<ColumnOrSuperColumn> results = client.get_slice(toByteBuffer(key_user_id), parent, predicate, ConsistencyLevel.ONE);
        for (ColumnOrSuperColumn result : results)
        {
            Column column = result.column;
            System.out.println(toString(column.name) + " -> " + toString(column.value));
        }

        tr.close();
    }
    
    public static ByteBuffer toByteBuffer(String value) 
    throws UnsupportedEncodingException
    {
        return ByteBuffer.wrap(value.getBytes("UTF-8"));
    }
          
    public static String toString(ByteBuffer buffer) 
    throws UnsupportedEncodingException
    {
        byte[] bytes = new byte[buffer.remaining()];
        buffer.get(bytes);
        return new String(bytes, "UTF-8");
    }
    
    /**
     * Create a connection and client for database access
     */
    public static Cassandra.Client getConnection(){
    	TTransport tr = new TFramedTransport(new TSocket("198.211.106.159", 9160));
        TProtocol proto = new TBinaryProtocol(tr);
        Cassandra.Client client = new Cassandra.Client(proto);
        try {
			tr.open();
		} catch (TTransportException e) {
			System.out.println("Failed to make a connection");
			e.printStackTrace();
		}
		return client;	
    }
    
    /**
     * Close the connection to the database
     * 
     * @param tr	connection
     */
    public static void closeConnection(TTransport tr){
    	tr.close();
    }
    
    /**
     * Inserts a row from the xml page provided by xmlSaxParser
     * 
     * @param key		User key ID
     * @param lad		Last Activity Date
     * @param id		ID
     * @param body		Text
     * @param score		Score
     */
    public static void insertRow(String key, String lad, String id, String body, String score){
    	
    }
}

